﻿/*
 * @author: wizardc
 */

using System;
using System.Collections.Generic;
using Dou.Utils;

namespace Dou.Operate
{
    /// <summary>
    /// 异步操作队列
    /// </summary>
    public class OperateQueue
    {
        private static readonly ObjectPool _queuePool = new ObjectPool(() => new Queue<OperateData>());
        
        private static int PrioritySort(int a, int b)
        {
            return b - a;
        }
        
        /// <summary>
        /// 并行执行允许的最大数量
        /// </summary>
        public int parallelMaxCount = 5;
        
        private Dictionary<Type, Type> _data2Executor = new Dictionary<Type, Type>();
        private Dictionary<Type, ObjectPool> _executorPoolMap = new Dictionary<Type, ObjectPool>();
        
        private List<int> _priorityList = new List<int>();
        private Dictionary<int, Queue<OperateData>> _priorityMap = new Dictionary<int, Queue<OperateData>>();
        private Dictionary<string, Queue<OperateData>> _keyMap = new Dictionary<string, Queue<OperateData>>();
        private HashSet<string> _operatingSet = new HashSet<string>();
        
        private int _parallelCount = 0;
        
        /// <summary>
        /// 绑定数据和执行器
        /// </summary>
        public void RegisterExecutor(Type dataType, Type executorType)
        {
            _data2Executor.Add(dataType, executorType);
        }
        
        private ObjectPool GetExecutorPool(Type dataType)
        {
            if (!_executorPoolMap.ContainsKey(dataType))
            {
                var executorType = _data2Executor[dataType];
                _executorPoolMap.Add(dataType, new ObjectPool(() => Activator.CreateInstance(executorType)));
            }
            return _executorPoolMap[dataType];
        }
        
        /// <summary>
        /// 添加一个操作对象进入队列
        /// </summary>
        public void AddOperate(OperateData operateData)
        {
            var priority = operateData.priority;
            if (!_priorityMap.ContainsKey(priority))
            {
                _priorityList.Add(priority);
                _priorityList.Sort(PrioritySort);
                _priorityMap.Add(priority, (Queue<OperateData>) _queuePool.Take());
            }
            var queue = _priorityMap[priority];
            queue.Enqueue(operateData);

            var key = operateData.key;
            if (!_keyMap.ContainsKey(key))
            {
                _keyMap.Add(key, (Queue<OperateData>) _queuePool.Take());
            }
            _keyMap[key].Enqueue(operateData);

            ExecuteNext();
        }

        private void ExecuteNext()
        {
            if (_parallelCount >= parallelMaxCount)
            {
                return;
            }

            OperateData operateData = null;

            for (int i = 0; i < _priorityList.Count; i++)
            {
                var priority = _priorityList[i];
                var queue = _priorityMap[priority];
                if (queue.Count > 0)
                {
                    operateData = queue.Dequeue();
                    break;
                }
                // 移除已经空掉的优先级记录
                _queuePool.Join(queue);
                _priorityMap.Remove(priority);
                _priorityList.RemoveAt(i);
                i--;
            }

            if (operateData == null)
            {
                return;
            }
            
            var key = operateData.key;
            if (_operatingSet.Contains(key)) // 当前处理的事务正在执行时直接尝试执行下一个处理
            {
                ExecuteNext();
            }
            else
            {
                _parallelCount++;
                _operatingSet.Add(key);

                var executorPool = GetExecutorPool(operateData.GetType());
                var executor = (IOperateExecutor) executorPool.Take();
                executor.Execute(operateData, OnExecuteComplete);
            }
        }

        private void OnExecuteComplete(IOperateExecutor executor, OperateData data, bool success, string error)
        {
            var executorPool = GetExecutorPool(data.GetType());
            executorPool.Join(executor);
            
            var key = data.key;
            
            _parallelCount--;
            _operatingSet.Remove(key);

            var dataQueue = _keyMap[key];
            foreach (var operateData in dataQueue)
            {
                operateData.onComplete(operateData, success, error);
            }
            dataQueue.Clear();
            _queuePool.Join(dataQueue);
            _keyMap.Remove(key);

            ExecuteNext(); // 尝试执行下一个处理
        }
    }
}
